package com.sunday.common.mq.rabbit.study.spring.study.e3_Receiving_Messages;

import com.rabbitmq.client.Channel;
import com.sunday.common.mq.rabbit.brave.annotation.TracingRabbitHandler;
import jakarta.validation.Valid;
import lombok.extern.slf4j.Slf4j;
import org.aopalliance.intercept.MethodInvocation;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor;
import org.springframework.amqp.rabbit.config.AbstractRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @Description: demo程序
 */
@Slf4j
@RestController
@RequestMapping("/test")
public class MQApi {

    @Autowired
    private RabbitListenerEndpointRegistry rabbitListenerEndpointRegistry;

    @Autowired
    private RabbitTemplate rabbitTemplate;

//    @Autowired
//    private TracingReceivePostProcessor tracingReceivePostProcessor;

    private final Random rand = new Random();

    public static final String listenerId = "queue_id";

    /**
     * ====================================== RabbitTemplate ================================
     */

    @GetMapping("/send")
    public void send() {
        ValidatedClass validatedClass = new ValidatedClass(rand.nextInt(10));
        log.info("send : {}", validatedClass);
        rabbitTemplate.convertAndSend(RabbitConfiguration.EXCHANGE_NAME, RabbitConfiguration.ROUTING_KEY_NAME, validatedClass);
    }

    @GetMapping("/batch")
    public void batch() {
        for (int i = 0; i < 10; i++) {
            ValidatedClass validatedClass = new ValidatedClass(rand.nextInt(10));
            log.info("send : {}", validatedClass);
            rabbitTemplate.convertAndSend(RabbitConfiguration.EXCHANGE_NAME, RabbitConfiguration.ROUTING_KEY_NAME, validatedClass);
        }
    }

    /**
     * ====================================== RabbitListenerEndpointRegistry ================================
     * <p>
     * {@link org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer}
     * {@link AbstractRabbitListenerContainerFactory#createListenerContainer(RabbitListenerEndpoint)}
     * {@link RabbitListenerEndpointRegistry#createListenerContainer(RabbitListenerEndpoint, RabbitListenerContainerFactory)}
     * {@link RabbitListenerAnnotationBeanPostProcessor#postProcessAfterInitialization(Object, String)}
     * <p>
     * https://docs.spring.io/spring-amqp/docs/3.0.4/reference/html/#preface
     * 当应用程序配置了单个 ，就像默认情况下使用 Spring 引导自动配置一样，当连接被代理阻止时，应用程序将停止工作。
     * 当它被经纪人阻止时，它的任何客户都会停止工作。
     * 如果我们在同一个应用程序中有生产者和消费者，当生产者阻止连接（因为代理上不再有资源）并且消费者无法释放它们（因为连接被阻止）时，
     * 我们最终可能会陷入死锁。 为了缓解此问题，我们建议再有一个具有相同选项的单独实例 - 一个用于生产者，一个用于使用者。
     * 对于在使用者线程上执行的事务生成者，不可能单独，因为它们应重用与使用者事务关联的。
     * CachingConnectionFactory CachingConnectionFactory CachingConnectionFactoryChannel
     */

//    @RabbitListener(
//            id = listenerId,
//            bindings = @QueueBinding(
//                    value = @Queue(value = RabbitConfiguration.QUEUE_NAME, durable = "true"),
//                    exchange = @Exchange(value = RabbitConfiguration.EXCHANGE_NAME, ignoreDeclarationExceptions = "true"),
//                    key = RabbitConfiguration.ROUTING_KEY_NAME),
//            concurrency = "1-5"
//    )
//    @RabbitListener(
//            id = listenerId,
//            bindings = @QueueBinding(
//                    value = @Queue(value = "auto.headers", autoDelete = "true",
//                            arguments = @Argument(name = "x-message-ttl", value = "10000",
//                                    type = "java.lang.Integer")),
//                    exchange = @Exchange(value = "auto.headers", type = ExchangeTypes.HEADERS, autoDelete = "true"),
//                    arguments = {
//                            @Argument(name = "x-match", value = "all"),
//                            @Argument(name = "thing1", value = "somevalue"),
//                            @Argument(name = "thing2")
//                    }),
//            concurrency = "1-5"
//    )

    /**
     * 从版本 2.0 开始，注释具有属性。 它支持 SpEL 表达式 （） 和属性占位符 （）。 其含义和允许的值取决于容器类型，如下所示：@RabbitListenerconcurrency#{…​}${…​}
     * 对于 ，该值必须是单个整数值，用于设置容器上的属性。DirectMessageListenerContainer consumersPerQueue
     * 对于 ，该值可以是单个整数值，用于设置容器上的属性，也可以具有 ， 其中 是属性和属性的形式。
     * SimpleRabbitListenerContainer concurrentConsumers m-n m concurrentConsumers n maxConcurrentConsumers
     */
//    @RabbitListener(
//            id = listenerId,
//            queues = {RabbitConfiguration.QUEUE_NAME},
//            concurrency = "1-5",
//            messageConverter = "gsonJsonMessageConverter"
//    )
//    public void taskPlanCheck(Message message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {
//        log.info("message ------>{}", message);
//        log.info("channel ------>{}", channel);
//        log.info("tag ------>{}", tag);
//        log.info("message ------>{}", message);
//    }

    /**
     * 在 Spring 框架中，@Payload 注解通常与消息传递机制一起使用，用于将请求或响应的消息有效负载（payload）映射到方法参数或返回值上。
     * 这个注解可以用在方法的参数上，表示该参数是用来接收请求消息体的数据，或者用在方法的返回值上，表示该返回值会作为响应消息的数据体。
     * 通常与 @RequestMapping 或 @MessageMapping 等其他注解一起使用，实现对应的 RESTful API 或 WebSocket 消息处理功能。
     */
//    @RabbitListener(
//            id = listenerId,
//            queues = RabbitConfiguration.QUEUE_NAME,
//            concurrency = "1-5",
//            errorHandler = "rabbitListenerErrorHandler"
////            ,messageConverter = "gsonJsonMessageConverter"
//    )
    public void taskPlanCheck(@Valid ValidatedClass validatedClass, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {
        log.info("channel ------>{}", channel);
        log.info("tag ------>{}", tag);
        log.info("message ------>{}", validatedClass);
        TimeUnit.SECONDS.sleep(1);
    }

    /**
     * https://docs.spring.io/spring-amqp/docs/current/reference/html/#receiving-batch
     *
     * @RabbitListener批处理 接收一批消息时，通常由容器执行去批处理，并且一次使用一条消息调用侦听器。 从版本 2.2 开始，您可以将侦听器容器工厂和侦听器配置为在一次调用中接收整个批处理，只需设置工厂的属性，并使方法有效负载参数为 a 或 ：
     * <p>
     * Message message = null;
     * if (methodInvocation.getArguments()[1] instanceof List) {
     * message = ((List<? extends Message>) methodInvocation.getArguments()[1]).get(0);
     * } else {
     * message = (Message) methodInvocation.getArguments()[1];
     * }
     * <p>
     * {@link brave.spring.rabbit.TracingRabbitListenerAdvice#invoke(MethodInvocation)}
     * 1. rabbit中引发了一个主要问题，其中参考kafka代码可知，批量获取的时候，执行结果如下
     * 发送记录 ->
     * INFO [655c7dc58fab987fffeba92ab065bae4,ffeba92ab065bae4] 3928 --- [io-16500-exec-1] c.s.c.m.r.s.s.s.e.MQApi [57] : send : {"bar":5}
     * INFO [655c7dc7fe30d34eb17b880a81a127c0,b17b880a81a127c0] 3928 --- [io-16500-exec-2] c.s.c.m.r.s.s.s.e.MQApi [57] : send : {"bar":6}
     * 拉取记录 -> 其中只能获取到其中的一个 message = ((List<? extends Message>) methodInvocation.getArguments()[1]).get(0);
     * 作为主要链路信息
     * traceparent -> 00-655c7dc58fab987fffeba92ab065bae4-eedfb6100b5fb200-01
     * traceparent -> 00-655c7dc7fe30d34eb17b880a81a127c0-b9911b918261d427-01
     * 程序执行利用的是aop能力
     * 2023-11-21 17:54:04.108  INFO [655c7dc58fab987fffeba92ab065bae4,61026e3ceea25b37] 3928 --- [     queue_id-1] c.s.c.m.r.s.s.s.e.MQApi [167] : message ------>(Body:'{"bar":5}' MessageProperties [headers={spring_listener_return_correlation=1e8c7341-4abc-4fce-9f3d-e59ec37c057d, __TypeId__=com.sunday.common.mq.rabbit.study.spring.study.e3_Receiving_Messages.ValidatedClass}, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=receiving_messages_exchange, receivedRoutingKey=receiving_messages_routing_key, deliveryTag=1, consumerTag=sunday-common-mq-rabbit_receiving_messages_queue, consumerQueue=receiving_messages_queue])
     * 2023-11-21 17:54:04.109  INFO [655c7dc58fab987fffeba92ab065bae4,61026e3ceea25b37] 3928 --- [     queue_id-1] c.s.c.m.r.s.s.s.e.MQApi [167] : message ------>(Body:'{"bar":6}' MessageProperties [headers={spring_listener_return_correlation=1e8c7341-4abc-4fce-9f3d-e59ec37c057d, traceparent=00-655c7dc7fe30d34eb17b880a81a127c0-b9911b918261d427-01, __TypeId__=com.sunday.common.mq.rabbit.study.spring.study.e3_Receiving_Messages.ValidatedClass}, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=receiving_messages_exchange, receivedRoutingKey=receiving_messages_routing_key, deliveryTag=2, consumerTag=sunday-common-mq-rabbit_receiving_messages_queue, consumerQueue=receiving_messages_queue])
     * 所以在获得结果是Batch的时候，第一个清空掉了 traceparent, 而第二个没有，全链路结果都是跟着第一个信息进行链路填充
     * 差异信息 : traceparent=00-655c7dc7fe30d34eb17b880a81a127c0-b9911b918261d427-01
     * 以上可知，如果是批量获取多个信息来进行链路的填充，链路的信息记录为首个信息带来的链路信息，
     * 也就是最终呈现上并不是完全正确，此处在批量进行的时候是存在坑的，但如果只是进行日志检索，问题也不是特别大，
     * 而且rabbit本身没有批量pull的能力，是由程序进行的处理，一般的使用基本都是单条执行的
     */
    @RabbitListener(
            id = listenerId,
            queues = RabbitConfiguration.QUEUE_NAME,
            concurrency = "1-5",
            errorHandler = "rabbitListenerErrorHandler",
// 特殊注意一下 当开启 batch以后，可能存在丢失问题
            containerFactory = "consumerBatchContainerFactory"
            , autoStartup = "false"
//            ,batch = "false"
    )
//    public void taskPlanCheck(@Valid List<ValidatedClass> messages, Channel channel) throws Exception {
    public void taskPlanCheck(@Valid List<Message> messages, Channel channel) throws Exception {
//    public void taskPlanCheck(@Valid Message message, Channel channel) throws Exception {
//        log.info("channel ------>{}", channel);
        messages.stream().forEach(message -> log.info("message ------>{}", message));
//        log.info("message ------>{}", message);
        TimeUnit.SECONDS.sleep(1);
    }

    /**
     * 第一个是用收到的原始的、未转换的 s 调用的。org.springframework.amqp.core.Message
     * 第二个是使用 S 调用的，其中包含转换的有效负载和映射的标头/属性。org.springframework.messaging.Message<?>
     * 第三个是使用转换后的有效负载调用的，无法访问标头/属性。
     * 您还可以添加参数，通常在使用确认模式时使用。 这对于第三个示例不是很有用，因为您无权访问该属性。ChannelMANUALdelivery_tag
     * Spring 引导为 和 提供了配置属性，但不为 提供配置属性。 从版本 3.0 开始，在容器工厂中设置为 也会设置为 。 当为 时，侦听器必须是批处理侦听器。consumerBatchEnabledbatchSizebatchListenerconsumerBatchEnabledtruebatchListenertrueconsumerBatchEnabledtrue
     * 从版本 3.0 开始，侦听器方法可以使用 或 。Collection<?>List<?>
     */

//    @RabbitListener(queues = RabbitConfiguration.QUEUE_NAME, containerFactory = "consumerBatchContainerFactory")
//    public void consumerBatch1(List<Message> amqpMessages) {
//        amqpMessages.stream().forEach(message -> log.info("message ------>{}", message));
//    }
//    @RabbitListener(queues = RabbitConfiguration.QUEUE_NAME, containerFactory = "consumerBatchContainerFactory")
//    public void consumerBatch2(List<org.springframework.messaging.Message<ValidatedClass>> messages) {
//        messages.stream().forEach(message -> log.info("message ------>{}", message));
//    }
//    @RabbitListener(queues = RabbitConfiguration.QUEUE_NAME, containerFactory = "consumerBatchContainerFactory")
//    public void consumerBatch3(List<ValidatedClass> strings) {
//        strings.stream().forEach(message -> log.info("message ------>{}", message));
//    }

    /**
     * 轮询消费者
     * 本身可用于轮询接收。 默认情况下，如果没有可用的消息，则立即返回。 没有阻塞。 从版本 1.5 开始，您可以设置 ，以毫秒为单位，接收方法最多阻止这么长时间，等待消息。 小于零的值表示无限期阻塞（或至少直到与代理的连接丢失）。 版本 1.6 引入了允许在每次调用时传递超时的方法变体。AmqpTemplateMessagenullreceiveTimeoutreceive
     * 由于接收操作会为每条消息创建一个新消息，因此此技术实际上并不适合高容量环境。 考虑对这些用例使用异步使用者或零。QueueingConsumerreceiveTimeout
     * 从版本 2.4.8 开始，使用非零超时时，可以指定传递到用于将使用者与通道相关联的方法中的参数。 例如：。basicConsumetemplate.addConsumerArg("x-priority", 10)
     */
    @GetMapping("receive")
    public void receive() throws UnsupportedEncodingException {
        Message message;
        while ((message = rabbitTemplate.receive(RabbitConfiguration.QUEUE_NAME)) != null) {
            log.info("message ------>{}", message);
            log.info("body ------>{}", new String(message.getBody(), "UTF-8"));
        }
    }

    @GetMapping("receiveAndConvert")
    public void receiveAndConvert() {
        Object time;
        while ((time = rabbitTemplate.receiveAndConvert(RabbitConfiguration.QUEUE_NAME)) != null) {
            log.info("message ------>{}", time);
        }
    }


    @GetMapping("receiveAndConvertType")
    public void receiveAndConvertType() {
        ParameterizedTypeReference<ValidatedClass> parameterizedTypeReference = new ParameterizedTypeReference<>() {
        };
        ValidatedClass time;
        while ((time = rabbitTemplate.receiveAndConvert(RabbitConfiguration.QUEUE_NAME, parameterizedTypeReference)) != null) {
            log.info("message ------>{}", time);
        }
    }

//    /**
//     * 自定义埋点
//     * 参考 {@link RabbitTemplate#doReceiveNoWait(String)}
//     */
//    @GetMapping("customReceive")
//    public void customReceive() {
//        tracingReceivePostProcessor.process(RabbitConfiguration.QUEUE_NAME,  message -> {
//            log.info("message ------>{}", message);
//            try {
//                TimeUnit.SECONDS.sleep(5);
//                log.info("body ------>{}", new String(message.getBody(), "UTF-8"));
//            } catch (Exception e) {
//                throw new RuntimeException(e);
//            }
//        });
//    }

    /**
     * Aop 自定义埋点
     * 参考 {@link RabbitTemplate#doReceiveNoWait(String)}
     */
    @TracingRabbitHandler
    @GetMapping("customAopReceive")
    public void customAopReceive() throws Exception {
        Message message;
        while ((message = rabbitTemplate.receive(RabbitConfiguration.QUEUE_NAME)) != null) {
            log.info("message ------>{}", message);
            log.info("body ------>{}", new String(message.getBody(), "UTF-8"));
            TimeUnit.SECONDS.sleep(1);
        }
    }

    /**
     * ====================================== RabbitListenerEndpointRegistry ================================
     */
    @GetMapping("stop")
    public void stop() {
        MessageListenerContainer listenerContainer = rabbitListenerEndpointRegistry.getListenerContainer(listenerId);
        log.info("{}", listenerContainer);
        listenerContainer.stop();
    }

    @GetMapping("start")
    public void start() {
        MessageListenerContainer listenerContainer = rabbitListenerEndpointRegistry.getListenerContainer(listenerId);
        log.info("{}", listenerContainer);
        listenerContainer.start();
    }

}